package com.gome.ocean.service.task;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.Callable;

import org.apache.commons.lang.StringUtils;
import org.quartz.JobExecutionException;

import com.alibaba.fastjson.JSON;
import com.gome.ocean.common.constants.DataxContants;
import com.gome.ocean.common.enums.State;
import com.gome.ocean.common.utils.AddressUtils;
import com.gome.ocean.common.utils.DataxHelper;
import com.gome.ocean.common.utils.RetryUtil;
import com.gome.ocean.common.utils.SpringContextUtil;
import com.gome.ocean.common.utils.TimeUtils;
import com.gome.ocean.dao.datax.JobDAO;
import com.gome.ocean.dao.datax.JobHistoryDAO;
import com.gome.ocean.dao.model.datax.JobBO;
import com.gome.ocean.dao.model.datax.JobHistoryBO;
import com.gome.ocean.service.utils.CmdToolkit;
import com.gome.ocean.service.zookeeper.service.ZkService;

/**
 * 调度任务
 * 
 * @author haqiaolong 2016年1月22日 下午4:06:09
 * @since 5.0.0
 */
public class TransferJobTask extends ScheduledJobTask {

    protected JobBO       jobBO;
    private JobDAO        jobDAO;
    private JobHistoryBO  jobHistoryBO;
    private JobHistoryDAO jobHistoryDAO;
    private String        taskResult;
    private ZkService     zkService;

    @Override
    public void initTask() throws JobExecutionException {
        if (this.jobDAO == null) this.jobDAO = SpringContextUtil.getBean("jobDAO");
        if (this.jobHistoryDAO == null) this.jobHistoryDAO = SpringContextUtil.getBean("jobHistoryDAO");
        if (this.zkService == null) this.zkService = SpringContextUtil.getBean("zkService");

        this.jobBO.setState(State.RUN.getValue());
        this.jobDAO.updateJobBO(jobBO);

        this.jobBO.setStartTime(null);
        this.jobBO.setEndTime(null);
        this.jobDAO.updateJobBO(jobBO);
    }

    @Override
    public void runTask() throws JobExecutionException {
        executeTask(); // 执行任务
        processLogInfo(); // 处理结果集
    }

    /**
     * 处理返回结果
     */
    private void processLogInfo() {
        if (StringUtils.isNotEmpty(taskResult)) {
            analyzeJobException(); // 分析异常
            analyzeJobPerf(); // 分析性能
        }
    }

    /**
     * 执行任务
     * @throws JobExecutionException 
     * @throws Exception 
     */
    private void executeTask() throws JobExecutionException {
        this.jobBO.setStartTime(new Date());
        this.jobDAO.updateJobBO(this.jobBO);
        try {
            this.taskResult = RetryUtil.executeWithRetry(new Callable<String>(){
                @Override
                public String call() throws Exception {
                    String result = CmdToolkit.executeCmd(DataxContants.getPythonCommondHttpPath(jobBO.getId(),
                        DataxContants.getServiceContextPath(zkService.getService())), new ConsoleCallBack());
                    return result;
                }
            }, 3, 1000L, true);
        } catch (Exception e) {
            throw new JobExecutionException(e);
        }
        this.jobBO.setEndTime(new Date());
    }

    class ConsoleCallBack implements CallBack {

        @Override
        public void callBack() {
            jobHistoryBO = new JobHistoryBO();
            jobHistoryBO.setTJobId(jobBO.getId());
            jobHistoryBO.setState(State.RUN.getValue());
            jobHistoryBO.setContext(jobBO.getContext());
            jobHistoryBO.setExecuteIp(AddressUtils.getHostIpAndPort());
            jobHistoryBO.setLogUrl(DataxHelper.getCurrentDataxLogPath(DataxContants.getFullFileName(jobBO.getId())));
            jobHistoryBO.setProjectId(jobBO.getProjectId());
            jobHistoryDAO.insertJobHistory(jobHistoryBO);
        }
    }

    @Override
    public boolean checkTask() throws JobExecutionException {
        if (this.jobBO == null) this.jobBO = (JobBO) context.getJobDetail().getJobDataMap().get("jobBO");
        State state = State.getState(this.jobBO.getState());
        if (State.RUN == state) {
            LOG.warn("任务正在运行！请检查!");
            return Boolean.FALSE; // 正在运行
        }
        return Boolean.TRUE;
    }

    /**
     * 分析任务性能
     * 
     * @param logInfo
     */
    private void analyzeJobPerf() {
        if (taskResult.contains("logStatistics info:")) {
            String subStr = StringUtils.substringAfterLast(taskResult, "logStatistics info:");
            if (StringUtils.isNotEmpty(subStr)) {
                subStr = StringUtils.substring(subStr, subStr.indexOf("{"), subStr.indexOf("}") + 1);
                @SuppressWarnings("unchecked")
                Map<String, Object> maps = JSON.parseObject(subStr, Map.class);
                if (maps.containsKey("startTime")) {
                    this.jobHistoryBO.setStartTime(TimeUtils.getDateTime(maps.get("startTime").toString()));
                }
                if (maps.containsKey("endTime")) {
                    this.jobHistoryBO.setEndTime(TimeUtils.getDateTime(maps.get("endTime").toString()));
                }
                if (maps.containsKey("byteSpeedPerSecond")) {
                    this.jobHistoryBO.setSpeedBytes(maps.get("byteSpeedPerSecond").toString());
                }
                if (maps.containsKey("recordSpeedPerSecond")) { // 记录写入速度
                    this.jobHistoryBO.setSpeedRecords(Long.valueOf(maps.get("recordSpeedPerSecond").toString()));
                }
                if (maps.containsKey("totalReadRecords")) { // 读出记录总数
                    this.jobHistoryBO.setTotalRecords(Long.valueOf(maps.get("totalReadRecords").toString()));
                }
                if (maps.containsKey("totalErrorRecords")) { // 错误数
                    this.jobHistoryBO.setErrorRecords(Long.valueOf(maps.get("totalErrorRecords").toString()));
                }
            }
        } else {
            this.jobHistoryBO.setStartTime(this.jobBO.getStartTime());
            this.jobHistoryBO.setEndTime(this.jobBO.getEndTime());
            this.jobHistoryBO.setSpeedBytes("0");
            this.jobHistoryBO.setSpeedRecords(0l);
            this.jobHistoryBO.setTotalRecords(0l);
            this.jobHistoryBO.setErrorRecords(0l);
        }
        this.jobDAO.updateJobBO(this.jobBO);

        this.jobHistoryBO.setSubmitUser(this.jobBO.getSubmitUser());
        this.jobHistoryBO.setSubmitTime(this.jobBO.getSubmitTime());
        this.jobHistoryBO.setSubmitIp(this.jobBO.getSubmitIp());
        this.jobHistoryBO.setPipelineName(this.jobBO.getPipelineName());
        this.jobHistoryBO.setState(this.jobBO.getState());
        this.jobHistoryBO.setCronExpression(this.jobBO.getCronExpression());
        this.jobHistoryDAO.updateJobHistoryById(jobHistoryBO);
        LOG.info("插入历史记录完成！jobId:" + this.jobBO.getId());
    }

    /**
     * 分析任务异常信息
     * 
     * @param logInfo
     */
    private void analyzeJobException() {
        if (taskResult.contains(DataxContants.DEFAULT_DATAX_ERROR)) {
            this.jobBO.setState(State.FAIL.getValue());

            this.jobHistoryBO.setState(State.FAIL.getValue());
            this.jobHistoryBO.setErrorMessage(StringUtils.substringAfterLast(taskResult,
                DataxContants.DEFAULT_DATAX_ERROR));
        } else {
            this.jobBO.setState(State.SUCCESS.getValue());
            this.jobHistoryBO.setState(State.SUCCESS.getValue());
        }
    }
}
